package com.xuecheng.messagesdk.service;

import com.xuecheng.messagesdk.model.po.MqMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @Author: LiJunXi
 * @Description:
 * @Date: 2024-08-31  10:02
 */
@Slf4j
@Component
public abstract class MessageProcessAbstract {

    @Autowired
    private MqMessageService mqMessageService;

    public abstract Boolean execute(MqMessage mqMessage);

    public void process(Integer shardIndex, Integer shardTotal, Integer total){
        try {
            // 取出该轮需要处理的消息列表
            List<MqMessage> mqMessageList = mqMessageService.getWaitTaskList(shardIndex, shardTotal, total);
            // 能够处理的最大消息数
            int size = mqMessageList.size();
            if (size == 0){
                return;
            }
            ExecutorService executorService = Executors.newFixedThreadPool(size);
            CountDownLatch countDownLatch = new CountDownLatch(size);
            log.info("开始执行消息任务");
            for (MqMessage mqMessage : mqMessageList) {
                try {
                    executorService.execute(()-> {
                        Boolean execute = execute(mqMessage);
                        if (!execute){
                            log.error("消息执行失败：{}", mqMessage);
                        }
                    });
                } catch (Exception e){
                    log.error("消息执行出错：{}", e.getMessage());
                } finally {
                    countDownLatch.countDown();
                }
            }
            //等待,给一个充裕的超时时间,防止无限等待，到达超时时间还没有处理完成则结束任务
            countDownLatch.await(30, TimeUnit.SECONDS);
            log.info("执行消息任务结束");

        } catch (Exception e){
            log.error("消息执行出错：{}", e.getMessage());
        }
    }

    public MqMessageService getMqMessageService(){
        return mqMessageService;
    }

}
